package com.jml.handler;

import com.jml.entity.DeliveryInfoEntity;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.apache.commons.lang3.StringUtils;

import java.util.*;


public class NettyMQServerHandler extends SimpleChannelInboundHandler<DeliveryInfoEntity> {
    /**
     * mq存放所有的队列
     */
    private static Map<String, Queue> queues = new HashMap<String, Queue>();

    /**
     * 存放我们消费者连接
     */
    private static Map<String, ChannelHandlerContext> ctxs = new HashMap<String, ChannelHandlerContext>();

    /**
     * Netty的服务器端接受客户端消息  MQ服务器端接收生产者和消费者的请求
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, DeliveryInfoEntity dInfo) throws Exception {
        String queueName = dInfo.getQueueName();
        if (StringUtils.isEmpty(queueName)) {
            return;
        }
        Boolean connType = dInfo.getConnType();
        // 如果是为true的情况下 为生产者角色
        if (connType) {
            // 处理生产者角色
            procucterService(queueName, dInfo.getMsg());
            return;
        }
        consumer(queueName, ctx);

    }

    private void procucterService(String queueName, String msg) {
        Queue queue = queues.get(queueName);
        if (queue == null) {
            // 如果队列不存在的情况下，就创建
            queue = new LinkedList();
            queues.put(queueName, queue);
        }
        // 将消息缓存到队列中
        queue.offer(msg);

        // 主动将消息推送给消费者
        ChannelHandlerContext ctx = ctxs.get(queueName);
        if (ctx != null) {
            ctx.writeAndFlush(queue.poll());
        }

    }

    /**
     * 消费和mq建立连接主动拉取消息
     */
    private void consumer(String queueName, ChannelHandlerContext ctx) {
        Queue queue = queues.get(queueName);
        if (queue == null) {
            return;
        }
        // 获取队列中消息
        Object poll = queue.poll();
        ctx.writeAndFlush(poll);

        //将消费者连接存放到集合中，方便主动推送消息给消费者
        ctxs.put(queueName, ctx);
    }

}